package com.springboot.kafka;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONObject;
import com.springboot.config.ProviderConfig;
import com.springboot.kafka.message.base.Body;
import com.springboot.kafka.message.base.KafkaMessage;
import com.springboot.kafka.message.user.UserKafka;

import lombok.extern.log4j.Log4j2;

/**
 * kafka消息发送者
 *
 * @author supu
 * @create 2019-01-28 17:43
 **/
@Log4j2
@Component
public class KafkaProvider {
    @Autowired
    private ProviderConfig providerConfig;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void userKafkaSend(KafkaMessage<UserKafka> kafkaMessage) {
        kafkaMessage.setMessageType("user");
        sendMessage(providerConfig.getString(ProviderConfig.KAFKA_TOPIC_USER), kafkaMessage);
    }

    private <T extends Body> void sendMessage(String topic, KafkaMessage<T> kafkaMessage) {
        if (!kafkaMessage.getBody().validate()) {
            log.info("topic:{} 发送的消息体：{}不符合要求，消息无法发送", topic, kafkaMessage.getBody());
            return;
        }

        String message = JSONObject.toJSONString(kafkaMessage);
        log.info("topic:{} 发送消息：{}", topic, message);

        kafkaTemplate.send(topic, message);
    }
}
